package com.wang.helloworld.storm.bolt;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class OtherMessageBolt  extends BaseRichBolt{
	private static final long serialVersionUID = 2652651175708547089L;
	Logger logger = LoggerFactory.getLogger(OtherMessageBolt.class);

	OutputCollector collector;
	Map map;
	TopologyContext context;
	
	public void execute(Tuple tuple) {
		System.out.println("Receive other message: " + tuple.getString(0) + "[message body:" + tuple.getString(1) + "]");

		// TODO: Filtering
		
		// Process chain..
		collector.emit(tuple, new Values(tuple.getString(0), tuple.getString(1)));

        collector.ack(tuple);
	}

	public void prepare(Map map, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		this.map = map;
		this.context = context;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("messageType","data"));
	}

}
